package liveData.migration.toPhoenix.task.run;

import com.alibaba.fastjson.JSON;
import liveData.migration.toPhoenix.task.db.PhoenixJdbc;
import liveData.migration.toPhoenix.task.entity.FlatMessage;
import liveData.migration.toPhoenix.task.util.ConstantUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * @program: LiveDataMigrationToPhoenix
 * @description:
 * @author: sunteng
 * @create: 2020-04-27 15:51
 **/
@Slf4j
public class VerifyResult {

    public static void dealDataStream() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(600000);
        env.getCheckpointConfig().setCheckpointTimeout(300000);
        env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
        env.setParallelism(3);

        log.info("##setting properties.....");
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConstantUtil.BOOTSTRAP_SERVER);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, ConstantUtil.GROUP_ID);
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, " ");


        FlinkKafkaConsumer011 flinkKafkaConsumer011 = new FlinkKafkaConsumer011(ConstantUtil.FROM_TOPIC, new SimpleStringSchema(), properties);

        flinkKafkaConsumer011.assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {
            @Override
            public long extractAscendingTimestamp(Object element) {
                return JSON.parseObject((String) element, FlatMessage.class).getEs();
            }
        });

        DataStream<String> stream = env
                .addSource(flinkKafkaConsumer011)
                .name("FROM-KAFKA");
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        if (stream != null) {
            try {
                stream
                        .map(a -> parseJson(a))
                        .name("PARSE")
                        .filter(new FilterFunction<FlatMessage>() {
                            @Override
                            public boolean filter(FlatMessage value) throws Exception {
                                return !value.getIsDdl();
                            }
                        })
                        .name("FILTER-DDL")
                        .addSink(new PhoenixJdbc())
                        .name("PHOENIX-SINK");

            } catch (Exception e) {
                log.error("##Exception is [{}]", e.getMessage());
            }
        }
        env.execute("migrationToPhoenix");
        log.info("##end consumer kafka from topic {}.............", ConstantUtil.FROM_TOPIC);

    }

    private static FlatMessage parseJson(String element) {

        return JSON.parseObject(element, FlatMessage.class);

    }

    public static void main(String[] args) {
        try {
            dealDataStream();
        } catch (Exception e) {
            log.error("migrationToPhoenix error message is [{}]", e.getMessage());
        }
    }
}
